package com.dynamicthreadpool.netty.client;

import com.dynamicthreadpool.netty.protocol.ThreadPoolControlMessage;
import com.dynamicthreadpool.netty.protocol.ThreadPoolMessageDecoder;
import com.dynamicthreadpool.netty.protocol.ThreadPoolMessageEncoder;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.channel.ChannelInboundHandlerAdapter;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

/**
 * 线程池控制客户端
 */
public class ThreadPoolControlClient {
    
    private final String host;
    private final int port;
    private final String accessToken;
    private ChannelHandlerContext context;
    private ThreadPoolControlMessage response;
    private CountDownLatch latch;
    
    public ThreadPoolControlClient(String host, int port) {
        this(host, port, null);
    }
    
    public ThreadPoolControlClient(String host, int port, String accessToken) {
        this.host = host;
        this.port = port;
        this.accessToken = accessToken;
    }
    
    public void connect() throws Exception {
        EventLoopGroup group = new NioEventLoopGroup();
        
        try {
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(group)
                    .channel(NioSocketChannel.class)
                    .option(ChannelOption.TCP_NODELAY, true)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ch.pipeline()
                                    .addLast(new ThreadPoolMessageDecoder())
                                    .addLast(new ThreadPoolMessageEncoder())
                                    .addLast(new ClientHandler());
                        }
                    });
            
            // 连接服务器
            ChannelFuture future = bootstrap.connect(host, port).sync();
            // 等待连接成功
            future.channel().closeFuture().awaitUninterruptibly(1000);
        } catch (Exception e) {
            System.err.println("Failed to connect to server: " + e.getMessage());
            throw e;
        }
    }
    
    public ThreadPoolControlMessage sendRequest(ThreadPoolControlMessage request) throws Exception {
        if (context == null) {
            throw new IllegalStateException("Client not connected");
        }
        
        // 重置响应和计数器
        this.response = null;
        this.latch = new CountDownLatch(1);
        
        // 设置访问令牌
        request.setAccessToken(accessToken);
        
        // 发送请求
        context.writeAndFlush(request);
        
        // 等待响应，超时5秒
        boolean received = latch.await(5, TimeUnit.SECONDS);
        
        if (!received) {
            throw new TimeoutException("No response received within timeout period");
        }
        
        return response;
    }
    
    public void disconnect() {
        if (context != null) {
            context.close();
            context = null;
        }
    }
    
    private class ClientHandler extends ChannelInboundHandlerAdapter {
        @Override
        public void channelActive(ChannelHandlerContext ctx) {
            // 保存上下文用于后续通信
            ThreadPoolControlClient.this.context = ctx;
        }
        
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) {
            if (msg instanceof ThreadPoolControlMessage) {
                // 接收到响应
                response = (ThreadPoolControlMessage) msg;
                // 释放等待的线程
                if (latch != null) {
                    latch.countDown();
                }
            }
        }
        
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
            cause.printStackTrace();
            ctx.close();
        }
    }
    
    public static void main(String[] args) {
        try {
            // 创建客户端，指定accessToken（如果服务器需要）
            String accessToken = "your_access_token_here"; // 根据服务器配置设置
            ThreadPoolControlClient client = new ThreadPoolControlClient("localhost", 8888, accessToken);
            client.connect();
            
            // 测试创建线程池
            testCreateThreadPool(client);
            
            // 测试获取线程池状态
            testStatus(client);
            
            // 测试更新线程池
            testUpdateThreadPool(client);
            
            // 测试关闭线程池
            testShutdownThreadPool(client);
            
            client.disconnect();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    
    private static void testCreateThreadPool(ThreadPoolControlClient client) throws Exception {
        System.out.println("\n--- Testing Create Thread Pool ---");
        
        ThreadPoolControlMessage request = new ThreadPoolControlMessage("create", "test-pool-1");
        Map<String, String> params = new HashMap<>();
        params.put("corethreads", "3");
        params.put("maxthreads", "5");
        params.put("queues", "10");
        params.put("keepalive", "60000");
        params.put("queueType", "LinkedBlockingQueue");
        params.put("rejectPolicy", "abort");
        params.put("threadname", "test-thread");
        request.setParameters(params);
        
        ThreadPoolControlMessage response = client.sendRequest(request);
        System.out.println("Response: " + response);
        System.out.println("Pool Status: " + response.getPoolStatus());
    }
    
    private static void testStatus(ThreadPoolControlClient client) throws Exception {
        System.out.println("\n--- Testing Get Thread Pool Status ---");
        
        ThreadPoolControlMessage request = new ThreadPoolControlMessage("status", "test-pool-1");
        ThreadPoolControlMessage response = client.sendRequest(request);
        
        System.out.println("Response: " + response);
        System.out.println("Pool Status: " + response.getPoolStatus());
    }
    
    private static void testUpdateThreadPool(ThreadPoolControlClient client) throws Exception {
        System.out.println("\n--- Testing Update Thread Pool ---");
        
        ThreadPoolControlMessage request = new ThreadPoolControlMessage("update", "test-pool-1");
        Map<String, String> params = new HashMap<>();
        params.put("corethreads", "5");
        params.put("maxthreads", "8");
        params.put("queues", "20");
        request.setParameters(params);
        
        ThreadPoolControlMessage response = client.sendRequest(request);
        System.out.println("Response: " + response);
        System.out.println("Updated Pool Status: " + response.getPoolStatus());
    }
    
    private static void testShutdownThreadPool(ThreadPoolControlClient client) throws Exception {
        System.out.println("\n--- Testing Shutdown Thread Pool ---");
        
        ThreadPoolControlMessage request = new ThreadPoolControlMessage("shutdown", "test-pool-1");
        ThreadPoolControlMessage response = client.sendRequest(request);
        
        System.out.println("Response: " + response);
    }
    
    // 自定义超时异常类
    public static class TimeoutException extends Exception {
        public TimeoutException(String message) {
            super(message);
        }
    }
}